#include "cgzCLua.h"
#include "cgzClickHouse.h"
#include "cgzCThreadPool.h"

static std::string DateTimeToString(long timestamp)
{
    time_t time = static_cast<time_t>(timestamp);
    struct tm tm;
    localtime_r(&time, &tm);
    std::ostringstream oss;
    oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S");
    return oss.str();
}

static void decode(cgzCLua *that, const cgzITask *iTask, cgzClickHouseExec *exce)
{
    // vector-会导致内存重新分配
    std::list<std::string> vstr;
    // 遍历结果集
    clargs2 crows = {};

    // 添加字段名
    clargs afield = {};

    switch (exce->m_cmd)
    {
    case cgzClickHouseCmd::SELECT:
    case cgzClickHouseCmd::EXPLAIN:
    case cgzClickHouseCmd::SHOW:
    {
        auto *pTask = static_cast<cgzClickHouseSelect *>(exce);
        if (nullptr == pTask->m_block)
        {
            break;
        }
        clickhouse::Block *block = pTask->m_block.get();
        // 填充列名称
        int colCnt = (int)block->GetColumnCount();
        for (int i = 0; i < colCnt; ++i)
        {
            const std::string &name = block->GetColumnName(i);
            afield.push_back(name.c_str());
        }

        // 填充行数据
        block->RefreshRowCount();
        int rowCnt = block->GetRowCount();
        for (int row = 0; row < rowCnt; row++)
        {
            crows.push_back(std::move(clargs()));
            clargs &arow = crows.back();
            for (int col = 0; col < colCnt; col++)
            {
                // 根据列类型填充数据
                auto column = block->operator[](col);
                switch (column->Type()->GetCode())
                {
                case clickhouse::Type::Void:
                {
                    arow.push_back((void *)nullptr);
                }
                break;
                case clickhouse::Type::Int8:
                {
                    long v = column->As<clickhouse::ColumnInt8>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::Int16:
                {
                    long v = column->As<clickhouse::ColumnInt16>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::Int32:
                {
                    long v = column->As<clickhouse::ColumnInt32>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::Int64:
                {
                    long v = column->As<clickhouse::ColumnInt64>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::UInt8:
                {
                    long v = column->As<clickhouse::ColumnUInt8>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::UInt16:
                {
                    long v = column->As<clickhouse::ColumnUInt16>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::UInt32:
                {
                    long v = column->As<clickhouse::ColumnUInt32>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::UInt64:
                {
                    long v = column->As<clickhouse::ColumnUInt64>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::Float32:
                {
                    double v = column->As<clickhouse::ColumnFloat32>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::Float64:
                {
                    double v = column->As<clickhouse::ColumnFloat64>()->At(row);
                    arow.push_back(v);
                }
                break;
                case clickhouse::Type::String:
                {
                    auto view = column->As<clickhouse::ColumnString>()->At(row);
                    vstr.emplace_back(view.data(), view.size());
                    // 取出最后一个元素压入
                    auto &it = vstr.back();
                    arow.emplace_back(it.c_str(), it.length());
                }
                break;
                case clickhouse::Type::FixedString:
                {
                    auto view = column->As<clickhouse::ColumnFixedString>()->At(row);
                    vstr.emplace_back(view.data(), view.size());
                    // 取出最后一个元素压入
                    auto &it = vstr.back();
                    arow.emplace_back(it.c_str(), it.length());
                }
                break;
                case clickhouse::Type::DateTime:
                {
                    long t = column->As<clickhouse::ColumnDateTime>()->At(row);
                    auto str = DateTimeToString(t);
                    vstr.emplace_back(std::move(str));
                    auto &it = vstr.back();
                    arow.push_back(it.c_str());
                    break;
                }
                case clickhouse::Type::Enum8:
                {
                    long v = column->As<clickhouse::ColumnEnum8>()->At(row);
                    auto view = column->Type()->As<clickhouse::EnumType>()->GetEnumName(v);
                    arow.push_back(view.data());
                }
                break;
                case clickhouse::Type::Enum16:
                {
                    long v = column->As<clickhouse::ColumnEnum16>()->At(row);
                    auto view = column->Type()->As<clickhouse::EnumType>()->GetEnumName(v);
                    arow.push_back(view.data());
                }
                break;
                default:
                {
                    arow.push_back((void *)nullptr);
                    break;
                }
                }
            }
        }
    }
    break;
    case cgzClickHouseCmd::INSERT:
    case cgzClickHouseCmd::INSERT_BLOCK:
    case cgzClickHouseCmd::UPDATE:
    case cgzClickHouseCmd::DELETE:
    case cgzClickHouseCmd::CREATE:
    case cgzClickHouseCmd::DROP:
    case cgzClickHouseCmd::ALTER:
    case cgzClickHouseCmd::USE:
    case cgzClickHouseCmd::TRUNCATE:
    {
    }
    break;
    }

    clargs args = {exce->m_cookid, afield, {crows, egzl_2array}};
    that->ocall(iTask->m_tref, "reply", args);
}

bool cgzCLua::run(const cgzITask *iTask, cgzClickHouseReply *reply)
{
    switch (reply->m_code)
    {
    case cgzClickHouseCode::AUTH_SUCCESS:
    {
        ocall(iTask->m_tref, "auth", {});
    }
    break;
    case cgzClickHouseCode::EXEC_REPLY:
    {
        auto exec = static_cast<cgzClickHouseExec *>(reply);
        if (exec->m_err)
        {
            ocall(iTask->m_tref, "reply", {exec->m_cookid, false});
        }
        else
        {
            ::decode(this, iTask, exec);
        }
    }
    break;
    default:
    {
        cgz_this_error("unknown cmd tskName:{}", iTask->getName());
    }
    break;
    }
    return true;
}

template <typename T, typename... Args>
void luaExecSqlTask(lua_State *L)
{
    // 将其转换为 cgzCLua 指针
    cgzCLua *that = cgzCLua::getThat(L);
    long cookid = lua_tointeger(L, 1);
    const char *name = lua_tostring(L, 2);
    const char *sql = lua_tostring(L, 3);
    void *node = malloc(sizeof(T));
    auto ask = new (node) T(sql, cookid);
    that->regTask(name, ask);
}

static int lickHouse_new(lua_State *L)
{
    // 将其转换为 cgzCLua 指针
    cgzCLua *that = cgzCLua::getThat(L);
    const char *name = lua_tostring(L, 1);
    const char *ip = lua_tostring(L, 2);
    int port = lua_tointeger(L, 3);
    const char *db = lua_tostring(L, 4);
    const char *user = lua_tostring(L, 5);
    const char *password = lua_tostring(L, 6);
    int tref = luaL_ref(L, LUA_REGISTRYINDEX);
    void *node = malloc(sizeof(cgzClickHouse));
    int cid = that->getId();
    auto cobj = new (node) cgzClickHouse(that->m_pool, name, ip, user, password, port, db, cid);
    cobj->m_tref = tref;
    return 0;
}

static int lickHouse_cblock(lua_State *L)
{
    // 1. 获取上下文
    cgzCLua *that = cgzCLua::getThat(L);
    if (!that->block)
    {
        that->block = std::make_unique<clickhouse::Block>();
    }
    else
    {
        that->block->Clear();
    }
    return 0;
}

#define BLOCK_NUMBER_APPEND(COLUMN, INT)       \
    {                                          \
        auto col = std::make_shared<COLUMN>(); \
        (block)->AppendColumn((colName), col); \
        lua_pushnil(L);                        \
        while (lua_next(L, 3))                 \
        {                                      \
            INT val = lua_tonumber(L, -1);     \
            col->Append(val);                  \
            lua_pop(L, 1);                     \
        }                                      \
    }

#define BLOCK_TIME64_APPEND(COLUMN, INT, P)     \
    {                                           \
        auto col = std::make_shared<COLUMN>(P); \
        (block)->AppendColumn((colName), col);  \
        lua_pushnil(L);                         \
        while (lua_next(L, 3))                  \
        {                                       \
            INT val = lua_tointeger(L, -1);     \
            col->Append(val);                   \
            lua_pop(L, 1);                      \
        }                                       \
    }

#define BLOCK_STRING_APPEND(COLUMN, STR)          \
    {                                             \
        auto col = std::make_shared<COLUMN>();    \
        (block)->AppendColumn((colName), col);    \
        lua_pushnil(L);                           \
        while (lua_next(L, 3))                    \
        {                                         \
            size_t len = 0;                       \
            STR val = lua_tolstring(L, -1, &len); \
            std::string str(val, len);            \
            col->Append(std::move(str));          \
            lua_pop(L, 1);                        \
        }                                         \
    }

static int lickHouse_column(lua_State *L)
{
    cgzCLua *that = cgzCLua::getThat(L);
    const char *colName = lua_tostring(L, 1);
    const int colType = lua_tointeger(L, 2);
    auto block = that->block.get();
    switch (colType)
    {
    case clickhouse::Type::Int8:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnInt8, int8_t)
        break;
    case clickhouse::Type::Int16:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnInt16, int16_t)
        break;
    case clickhouse::Type::Int32:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnInt32, int32_t)
        break;
    case clickhouse::Type::Int64:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnInt64, int64_t)
        break;
    case clickhouse::Type::UInt8:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnUInt8, uint8_t)
        break;
    case clickhouse::Type::UInt16:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnUInt16, uint16_t)
        break;
    case clickhouse::Type::UInt32:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnUInt32, uint32_t)
        break;
    case clickhouse::Type::UInt64:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnUInt64, uint64_t)
        break;
    case clickhouse::Type::Float32:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnFloat32, double)
        break;
    case clickhouse::Type::Float64:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnFloat64, double)
        break;
    case clickhouse::Type::String:
        BLOCK_STRING_APPEND(clickhouse::ColumnString, const char *)
        break;
    case clickhouse::Type::DateTime:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnDateTime, std::time_t)
        break;
    case clickhouse::Type::Date:
        BLOCK_NUMBER_APPEND(clickhouse::ColumnDate, std::time_t)
        break;
    case clickhouse::Type::DateTime64:
        BLOCK_TIME64_APPEND(clickhouse::ColumnDateTime64, int64_t, 3)
        break;
    default:
        break;
    }
    return 0;
}

static int lickHouse_insertBlock(lua_State *L)
{
    // 2. 获取上下文
    cgzCLua *that = cgzCLua::getThat(L);

    const long cook = lua_tointeger(L, 1);
    const char *name = lua_tostring(L, 2);
    if (!name || strlen(name) == 0)
    {
        return luaL_error(L, "Empty task name");
    }

    // 3. 提取表名
    const char *table = lua_tostring(L, 3);
    if (!table || strlen(table) == 0)
    {
        return luaL_error(L, "Empty table name");
    }

    that->block->RefreshRowCount();

    // 5. 创建任务对象
    void *const memory = malloc(sizeof(cgzClickHouseInsertBlock));
    auto *const task = new (memory) cgzClickHouseInsertBlock(table, std::move(that->block), cook);
    that->regTask(name, task);
    return 0;
}

static int lickHouse_insert(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseInsert>(L);
    return 0;
}

static int lickHouse_update(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseUpdate>(L);
    return 0;
}

static int lickHouse_delete(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseDelete>(L);
    return 0;
}

static int lickHouse_select(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseSelect>(L);
    return 0;
}

static int lickHouse_create(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseCreate>(L);
    return 0;
}

static int lickHouse_drop(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseDrop>(L);
    return 0;
}

static int lickHouse_alter(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseAlter>(L);
    return 0;
}

static int lickHouse_truncate(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseTruncate>(L);
    return 0;
}

static int lickHouse_explain(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseExplain>(L);
    return 0;
}

static int lickHouse_show(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseShow>(L);
    return 0;
}

static int lickHouse_use(lua_State *L)
{
    luaExecSqlTask<cgzClickHouseUse>(L);
    return 0;
}

static int lickHouse_callback(lua_State *L)
{
    cgzCLua *that = cgzCLua::getThat(L);
    long cookid = lua_tointeger(L, 1);
    auto task = (cgzClickHouseExec *)that->m_retTask.back();
    assert(task->m_cookid == cookid);
    task->m_ret = true;
    return 0;
}

int cgzCLua::luaopen_clickHouse(lua_State *L)
{
    static const luaL_Reg otherlib[] = {
        {"new", lickHouse_new},
        {"cblock", lickHouse_cblock},
        {"column", lickHouse_column},
        {"insert", lickHouse_insert},
        {"update", lickHouse_update},
        {"delete", lickHouse_delete},
        {"select", lickHouse_select},
        {"create", lickHouse_create},
        {"drop", lickHouse_drop},
        {"alter", lickHouse_alter},
        {"truncate", lickHouse_truncate},
        {"explain", lickHouse_explain},
        {"show", lickHouse_show},
        {"use", lickHouse_use},
        {"insertBlock", lickHouse_insertBlock},
        {"callback", lickHouse_callback},
        {NULL, NULL}};

    luaL_newlib(L, otherlib);
    return 1;
}
